package com.bw.app.dws;

import com.bw.bean.ProvinceStats;
import com.bw.utils.ClickHouseUtil;
import com.bw.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class ProvinceStatsSqlApp {
    public static void main(String[] args) throws Exception {
        // 1、创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        String groupId = "province_stats_2105b";
        String orderWideTopic = "dwm_order_wide";
        // 2、创建表环境

        // 3、建表，把Kafka动态读进该表中
        tableEnv.executeSql("CREATE TABLE ORDER_WIDE (" +
                "  `province_id` BIGINT," +
                "  `province_name` STRING," +
                "  `province_area_code` STRING," +
                "  `province_iso_code` STRING," +
                "  `province_3166_2_code` STRING," +
                "  `order_id` STRING," +
                "  `split_total_amount` DOUBLE," +
                "  `create_time` STRING," +
                "  `rt` as TO_TIMESTAMP(create_time), "+
                "  WATERMARK FOR rt AS rt - INTERVAL '1' SECOND )" +
                " WITH (" + MyKafkaUtil.getKafkaDDL(orderWideTopic,groupId)+ ")");
        // 4、查询SQL聚合操作
        /*
            select
                 province_id,province_name,province_area_code,province_3166_2_code,
                 sum(order_amount),
                 count(distinct(order_id))
                 ts uninx_timestamp
            from ORDER_WIDE group  by province_id,province_name,province_area_code,province_3166_2_code
         */
        Table table = tableEnv.sqlQuery("select " +
                "    DATE_FORMAT(TUMBLE_START(rt, INTERVAL '10' SECOND), 'yyyy-MM-dd HH:mm:ss') stt, " +
                "    DATE_FORMAT(TUMBLE_END(rt, INTERVAL '10' SECOND), 'yyyy-MM-dd HH:mm:ss') edt, " +
                "    province_id, " +
                "    province_name, " +
                "    province_area_code, " +
                "    province_iso_code, " +
                "    province_3166_2_code, " +
                "    count(distinct order_id) order_count, " +
                "    sum(split_total_amount) order_amount, " +
                "    UNIX_TIMESTAMP()*1000 ts " +
                "from " +
                "    ORDER_WIDE " +
                "group by " +
                "    province_id, " +
                "    province_name, " +
                "    province_area_code, " +
                "    province_iso_code, " +
                "    province_3166_2_code, " +
                "    TUMBLE(rt, INTERVAL '10' SECOND)");// 10秒开窗

        // 5. 把表转成流
        DataStream<ProvinceStats> provinceStatsDataStream = tableEnv.toAppendStream(table, ProvinceStats.class);
        // 6、写入CK中
        provinceStatsDataStream.print();
        provinceStatsDataStream.addSink(ClickHouseUtil.sink("insert into province_stats_2105b values(?,?,?,?,?,?,?,?,?,?)"));
        // 7、 启动任务
        env.execute("ProvinceStatsSqlApp");
    }
}
